# 导入URL编码解析库，用于处理HTTP请求参数的编码
import urllib.parse
# 导入HTTP请求库，用于发送网络请求
import requests
# 导入JSON处理库，用于解析API返回的JSON数据
import json
# 导入pandas数据分析库，用于数据处理和CSV文件操作
import pandas as pd
# 导入日期时间库，用于处理时间相关操作
from datetime import datetime, timedelta
# 导入时间库，用于设置延时等待
import time
# 导入操作系统库，用于文件路径和系统命令操作
import os
# 导入日志库，用于记录程序运行状态和错误信息
import logging
# 导入系统库，用于获取系统信息和标准输出
import sys
# 导入Kafka生产者，用于将数据实时发送到Kafka消息队列
from kafka import KafkaProducer
# 导入默认字典，用于创建带默认值的字典
from collections import defaultdict


# 配置日志系统的函数
def setup_logging_with_utf8():
    """配置UTF-8编码的日志系统"""
    # 清除现有的所有日志处理器，避免重复输出
    for handler in logging.root.handlers[:]:
        logging.root.removeHandler(handler)

    # 配置基础日志设置
    logging.basicConfig(
        level=logging.INFO,  # 设置日志级别为INFO，记录一般信息
        format='%(asctime)s - %(levelname)s - %(message)s',  # 设置日志格式：时间-级别-消息
        handlers=[
            # 文件处理器：将日志写入文件
            logging.FileHandler(
                'xinfadi_pagination.log',  # 日志文件名
                mode='a',  # 追加模式，不覆盖已有日志
                encoding='utf-8'  # 使用UTF-8编码，支持中文
            ),
            # 控制台处理器：将日志输出到控制台
            logging.StreamHandler(stream=sys.stdout)
        ]
    )

    # 如果是Windows系统，设置控制台编码为UTF-8
    if sys.platform.startswith('win'):
        os.system('chcp 65001')  # Windows命令：设置代码页为UTF-8


# 调用日志配置函数，初始化日志系统
setup_logging_with_utf8()


# 定义新发地分页爬虫类
class XinfadiPaginationCrawler:
    def __init__(self, max_pages=100, page_delay=2):
        """
        初始化分页爬虫
        :param max_pages: 最大爬取页数
        :param page_delay: 页面间延时（秒）
        """
        # 存储最大爬取页数
        self.max_pages = max_pages
        # 存储页面间延时时间，防止请求过于频繁被封IP
        self.page_delay = page_delay
        # 新发地价格数据API接口地址
        self.url = 'http://www.xinfadi.com.cn/getPriceData.html'

        # 设置HTTP请求头，模拟真实浏览器访问
        self.headers = {
            'Accept': '*/*',  # 接受所有类型的响应
            'Accept-Encoding': 'gzip, deflate',  # 支持压缩编码
            'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',  # 语言优先级：中文>英文
            'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8',  # POST数据格式
            'Host': 'www.xinfadi.com.cn',  # 目标主机
            'Origin': 'http://www.xinfadi.com.cn',  # 请求来源
            'Referer': 'http://www.xinfadi.com.cn/priceDetail.html',  # 引用页面
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',  # 浏览器标识
            'X-Requested-With': 'XMLHttpRequest',  # 标识为AJAX请求
        }

        # 创建数据存储目录
        self.data_dir = 'pagination_data'  # 数据存储目录名
        if not os.path.exists(self.data_dir):  # 如果目录不存在
            os.makedirs(self.data_dir)  # 创建目录

        # 初始化统计信息字典，用于记录爬取过程的各项指标
        self.stats = {
            'total_pages': 0,  # 总页数
            'total_records': 0,  # 总记录数
            'successful_pages': 0,  # 成功爬取的页数
            'failed_pages': 0,  # 失败的页数
            'start_time': None,  # 开始时间
            'end_time': None  # 结束时间
        }

        # 设置Kafka生产者，用于实时数据传输
        self.setup_kafka_producer()

    def setup_kafka_producer(self):
        """设置Kafka生产者"""
        try:
            # 创建Kafka生产者实例
            self.kafka_producer = KafkaProducer(
                bootstrap_servers=['192.168.93.201:9092'],  # Kafka服务器地址和端口
                # 值序列化器：将Python对象转换为JSON字符串，再编码为UTF-8字节
                value_serializer=lambda x: json.dumps(x, ensure_ascii=False).encode('utf-8'),
                # 键序列化器：处理字符串和字节类型的键
                key_serializer=lambda x: x if isinstance(x, bytes) else x.encode('utf-8'),
                api_version=(0, 10, 1),  # Kafka API版本
                retries=3,  # 重试次数
                batch_size=16384,  # 批处理大小（字节）
                linger_ms=10,  # 等待时间（毫秒），用于批量发送
                buffer_memory=33554432,  # 缓冲区大小（32MB）
                max_block_ms=60000,  # 最大阻塞时间（60秒）
                request_timeout_ms=30000,  # 请求超时时间（30秒）
                acks='all'  # 确认级别：等待所有副本确认
            )

            # 设置Kafka主题名称
            self.kafka_topic = 'agriculture_pagination_data'
            # 记录成功信息
            logging.info("Kafka生产者初始化成功")

        except Exception as e:
            # 记录错误信息
            logging.error(f"Kafka生产者初始化失败: {e}")
            # 设置为None表示Kafka不可用
            self.kafka_producer = None

    def fetch_page_data(self, page_number, product_name="", start_date="", end_date=""):
        """
        爬取指定页面的数据
        :param page_number: 页码
        :param product_name: 产品名称（可选）
        :param start_date: 开始日期
        :param end_date: 结束日期
        :return: 页面数据列表
        """
        # 设置默认日期范围（最近30天）
        if not end_date:  # 如果没有指定结束日期
            # 使用当前日期作为结束日期
            end_date = datetime.now().strftime('%Y/%m/%d')
        if not start_date:  # 如果没有指定开始日期
            # 使用30天前的日期作为开始日期
            start_date = (datetime.now() - timedelta(days=30)).strftime('%Y/%m/%d')

        # 构造POST请求的数据参数
        data = {
            'limit': '20',  # 每页显示20条数据
            'current': str(page_number),  # 当前页码
            'pubDateStartTime': start_date,  # 发布开始时间
            'pubDateEndTime': end_date,  # 发布结束时间
            'prodPcatid': '',  # 产品父分类ID（空表示所有）
            'prodCatid': '',  # 产品分类ID（空表示所有）
            'prodName': product_name,  # 产品名称
        }

        try:
            # 记录当前爬取的页面信息
            logging.info(f"正在爬取第{page_number}页数据...")

            # 发送POST请求到新发地API
            response = requests.post(
                self.url,  # 请求URL
                headers=self.headers,  # 请求头
                data=urllib.parse.urlencode(data),  # 将数据字典编码为URL格式
                timeout=15  # 设置15秒超时
            )

            # 检查HTTP响应状态码
            if response.status_code == 200:  # 请求成功
                try:
                    # 解析JSON响应数据
                    json_data = response.json()

                    # 检查响应中是否包含数据列表
                    if 'list' in json_data and json_data['list']:
                        page_data = []  # 初始化页面数据列表
                        items = json_data['list']  # 获取数据项

                        # 处理不同的数据结构格式
                        if isinstance(items, dict) and 'data' in items:
                            # 如果items是字典且包含data字段
                            items = items['data']
                        elif isinstance(items, list):
                            # 如果items已经是列表，直接使用
                            pass
                        else:
                            # 如果是未知的数据结构
                            logging.warning(f"第{page_number}页: 未知的数据结构")
                            return []

                        # 遍历处理每个数据项
                        for item in items:
                            # 构造标准化的数据记录
                            record = {
                                'page_number': page_number,  # 页码
                                'product_name': item.get('prodName', ''),  # 产品名称
                                'low_price': float(item.get('lowPrice', 0)),  # 最低价格
                                'high_price': float(item.get('highPrice', 0)),  # 最高价格
                                'avg_price': float(item.get('avgPrice', 0)),  # 平均价格
                                'place': item.get('place', ''),  # 产地
                                'spec': item.get('specInfo', ''),  # 规格信息
                                'unit': item.get('unitInfo', '元/公斤'),  # 计量单位
                                'pub_date': item.get('pubDate', ''),  # 发布日期
                                'crawl_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),  # 爬取时间
                                'timestamp': datetime.now().timestamp()  # 时间戳
                            }
                            # 将记录添加到页面数据列表
                            page_data.append(record)

                        # 记录成功获取的数据条数
                        logging.info(f"第{page_number}页: 成功获取{len(page_data)}条数据")
                        return page_data  # 返回页面数据
                    else:
                        # 如果没有数据返回
                        logging.warning(f"第{page_number}页: 无数据返回")
                        return []

                except json.JSONDecodeError as e:
                    # JSON解析失败
                    logging.error(f"第{page_number}页: JSON解析失败 - {e}")
                    return []
            else:
                # HTTP请求失败
                logging.error(f"第{page_number}页: 请求失败，状态码: {response.status_code}")
                return []

        except Exception as e:
            # 捕获所有其他异常
            logging.error(f"第{page_number}页: 请求异常 - {e}")
            return []

    def send_page_to_kafka(self, page_data, page_number):
        """发送页面数据到Kafka"""
        # 检查Kafka生产者是否可用且有数据要发送
        if not self.kafka_producer or not page_data:
            return

        try:
            # 为每条记录生成批次ID，包含页码和时间戳
            batch_id = f"page_{page_number}_{datetime.now().strftime('%Y%m%d_%H%M%S')}"

            # 遍历页面数据中的每条记录
            for record in page_data:
                # 复制原始记录，避免修改原数据
                enriched_record = record.copy()
                # 为记录添加额外的元数据信息
                enriched_record.update({
                    'batch_id': batch_id,  # 批次ID
                    'kafka_send_time': datetime.now().isoformat(),  # Kafka发送时间
                    'data_source': 'xinfadi_pagination_crawler',  # 数据来源标识
                    # 生成唯一记录ID
                    'record_id': f"page_{page_number}_{record['product_name']}_{record['timestamp']}"
                })

                # 发送记录到Kafka主题
                future = self.kafka_producer.send(
                    self.kafka_topic,  # 主题名称
                    value=enriched_record,  # 消息值（记录数据）
                    key=f"page_{page_number}".encode('utf-8')  # 消息键（页码）
                )

            # 强制发送所有缓冲的消息
            self.kafka_producer.flush()
            # 记录发送成功信息
            logging.info(f"第{page_number}页数据已发送到Kafka")

        except Exception as e:
            # 记录Kafka发送失败的错误
            logging.error(f"第{page_number}页Kafka发送失败: {e}")

    def crawl_by_pagination(self, product_name="", start_page=1, end_page=None):
        """
        按分页爬取数据
        :param product_name: 指定产品名称（空则爬取所有）
        :param start_page: 起始页码
        :param end_page: 结束页码（None则使用max_pages）
        """
        # 如果没有指定结束页码，使用初始化时设定的最大页数
        if end_page is None:
            end_page = self.max_pages

        # 记录爬取开始时间
        self.stats['start_time'] = datetime.now()
        # 初始化所有数据的列表
        all_data = []

        # 记录爬取范围信息
        logging.info(f"开始分页爬取: 第{start_page}页到第{end_page}页")
        if product_name:  # 如果指定了产品名称
            logging.info(f"指定产品: {product_name}")

        # 循环爬取指定范围内的每一页
        for page_num in range(start_page, end_page + 1):
            try:
                # 爬取当前页面的数据
                page_data = self.fetch_page_data(page_num, product_name)

                if page_data:  # 如果成功获取到数据
                    # 将页面数据添加到总数据列表
                    all_data.extend(page_data)
                    # 成功页数计数器加1
                    self.stats['successful_pages'] += 1
                    # 总记录数增加
                    self.stats['total_records'] += len(page_data)

                    # 保存单页数据到CSV文件
                    self.save_page_data(page_data, page_num)

                    # 发送数据到Kafka消息队列
                    self.send_page_to_kafka(page_data, page_num)

                else:  # 如果没有获取到数据
                    # 失败页数计数器加1
                    self.stats['failed_pages'] += 1
                    # 记录警告信息
                    logging.warning(f"第{page_num}页无数据，可能已到达最后一页")

                    # 检查是否连续3页无数据，如果是则停止爬取
                    if page_num - start_page >= 2:  # 如果已经爬取了至少3页
                        empty_pages = 0  # 空页面计数器
                        # 检查最近3页的数据情况
                        for check_page in range(max(start_page, page_num - 2), page_num + 1):
                            # 如果在all_data中找不到该页的数据
                            if not any(d['page_number'] == check_page for d in all_data):
                                empty_pages += 1

                        # 如果连续3页都没有数据
                        if empty_pages >= 3:
                            logging.info("连续3页无数据，停止爬取")
                            break  # 跳出循环，停止爬取

                # 总页数计数器加1（无论成功还是失败）
                self.stats['total_pages'] += 1

                # 页面间延时，避免请求过于频繁
                if page_num < end_page:  # 如果不是最后一页
                    logging.info(f"等待{self.page_delay}秒后爬取下一页...")
                    time.sleep(self.page_delay)  # 等待指定时间

            except KeyboardInterrupt:
                # 用户按Ctrl+C中断程序
                logging.info("用户中断爬取")
                break  # 跳出循环
            except Exception as e:
                # 捕获其他所有异常
                logging.error(f"第{page_num}页爬取异常: {e}")
                # 失败页数计数器加1
                self.stats['failed_pages'] += 1
                continue  # 继续下一页的爬取

        # 记录爬取结束时间
        self.stats['end_time'] = datetime.now()

        # 如果有数据，保存汇总数据
        if all_data:
            self.save_summary_data(all_data, product_name)

        # 输出爬取统计信息
        self.print_crawl_stats()

        # 返回所有爬取的数据
        return all_data

    def save_page_data(self, page_data, page_number):
        """保存单页数据"""
        # 如果没有数据，直接返回
        if not page_data:
            return

        # 生成带时间戳的文件名
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = os.path.join(self.data_dir, f'page_{page_number}_{timestamp}.csv')

        # 将数据转换为DataFrame并保存为CSV
        df = pd.DataFrame(page_data)
        df.to_csv(filename, index=False, encoding='utf-8-sig')  # 使用UTF-8-BOM编码，Excel兼容
        # 记录调试信息
        logging.debug(f"第{page_number}页数据已保存到: {filename}")

    def save_summary_data(self, all_data, product_name=""):
        """保存汇总数据"""
        # 如果没有数据，直接返回
        if not all_data:
            return

        # 生成带时间戳的汇总文件名
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        # 根据是否指定产品名称生成不同的文件名后缀
        product_suffix = f"_{product_name}" if product_name else "_all"
        filename = os.path.join(self.data_dir, f'summary{product_suffix}_{timestamp}.csv')

        # 将所有数据转换为DataFrame
        df = pd.DataFrame(all_data)

        # 按页码和爬取时间排序
        df = df.sort_values(['page_number', 'crawl_time'])

        # 去除重复数据，保留最后一次爬取的记录
        df = df.drop_duplicates(
            subset=['product_name', 'pub_date', 'avg_price', 'place', 'spec'],  # 去重的字段
            keep='last'  # 保留最后一条重复记录
        )

        # 保存去重后的汇总数据
        df.to_csv(filename, index=False, encoding='utf-8-sig')
        logging.info(f"汇总数据已保存到: {filename}")

        # 生成数据分析报告
        self.generate_analysis_report(df, filename.replace('.csv', '_analysis.txt'))

    def generate_analysis_report(self, df, filename):
        """生成数据分析报告"""
        try:
            # 以UTF-8编码写入文本文件
            with open(filename, 'w', encoding='utf-8') as f:
                # 写入报告标题
                f.write("=== 新发地菜价爬取分析报告 ===\n\n")
                # 写入爬取时间
                f.write(f"爬取时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
                # 写入总记录数
                f.write(f"总记录数: {len(df)}\n")
                # 写入产品种类数（去重后的产品名称数量）
                f.write(f"产品种类: {df['product_name'].nunique()}\n")
                # 写入价格范围（最低价到最高价）
                f.write(f"价格范围: {df['avg_price'].min():.2f} - {df['avg_price'].max():.2f} 元\n")
                # 写入平均价格
                f.write(f"平均价格: {df['avg_price'].mean():.2f} 元\n\n")

                # 按产品统计价格信息
                f.write("=== 产品价格统计 ===\n")
                # 按产品名称分组，计算平均价格的统计信息
                product_stats = df.groupby('product_name')['avg_price'].agg(['count', 'mean', 'min', 'max']).round(2)
                # 写入产品统计表格
                f.write(product_stats.to_string())
                f.write("\n\n")

                # 按产地统计
                f.write("=== 产地统计 ===\n")
                # 统计各产地的产品数量，取前10名
                place_stats = df['place'].value_counts().head(10)
                # 写入产地统计
                f.write(place_stats.to_string())

            # 记录报告生成成功信息
            logging.info(f"分析报告已生成: {filename}")

        except Exception as e:
            # 记录报告生成失败的错误
            logging.error(f"生成分析报告失败: {e}")

    def print_crawl_stats(self):
        """打印爬取统计信息"""
        # 计算总耗时
        duration = self.stats['end_time'] - self.stats['start_time']

        # 输出各项统计信息
        logging.info("=== 爬取统计信息 ===")
        logging.info(f"总页数: {self.stats['total_pages']}")
        logging.info(f"成功页数: {self.stats['successful_pages']}")
        logging.info(f"失败页数: {self.stats['failed_pages']}")
        logging.info(f"总记录数: {self.stats['total_records']}")
        logging.info(f"耗时: {duration}")
        # 计算平均每页记录数，避免除零错误
        logging.info(f"平均每页记录数: {self.stats['total_records'] / max(1, self.stats['successful_pages']):.1f}")

    def crawl_specific_products(self, product_list, max_pages_per_product=50):
        """
        爬取指定产品列表
        :param product_list: 产品名称列表
        :param max_pages_per_product: 每个产品最大爬取页数
        """
        # 初始化结果字典，用于存储每个产品的爬取结果
        all_results = {}

        # 遍历产品列表，逐个爬取
        for product in product_list:
            logging.info(f"开始爬取产品: {product}")

            # 调用分页爬取方法，爬取当前产品的数据
            product_data = self.crawl_by_pagination(
                product_name=product,  # 指定产品名称
                start_page=1,  # 从第1页开始
                end_page=max_pages_per_product  # 爬取指定的最大页数
            )

            # 将爬取结果存储到结果字典中
            all_results[product] = product_data

            # 产品间延时，避免过于频繁的请求
            if product != product_list[-1]:  # 如果不是最后一个产品
                logging.info(f"等待{self.page_delay * 2}秒后爬取下一个产品...")
                time.sleep(self.page_delay * 2)  # 等待时间是页面延时的2倍

        # 返回所有产品的爬取结果
        return all_results

    def close(self):
        """关闭资源"""
        # 检查是否有Kafka生产者需要关闭
        if hasattr(self, 'kafka_producer') and self.kafka_producer:
            try:
                # 强制发送所有缓冲的消息
                self.kafka_producer.flush()
                # 关闭Kafka生产者连接
                self.kafka_producer.close()
                logging.info("Kafka生产者已关闭")
            except Exception as e:
                # 记录关闭失败的错误
                logging.error(f"关闭Kafka生产者失败: {e}")


# 主函数，程序入口
def main():
    # 创建分页爬虫实例，设置最大100页，页面间延时2秒
    crawler = XinfadiPaginationCrawler(max_pages=100, page_delay=2)

    try:
        # 显示菜单选项
        print("新发地菜价分页爬虫")
        print("1. 爬取所有产品（分页）")
        print("2. 爬取指定产品")
        print("3. 爬取热门蔬菜")

        # 获取用户选择，去除首尾空格
        choice = input("请选择爬取方式 (1-3): ").strip()

        if choice == "1":
            # 选项1：爬取所有产品，最多100页
            crawler.crawl_by_pagination(start_page=1, end_page=100)

        elif choice == "2":
            # 选项2：爬取指定产品
            # 获取用户输入的产品名称
            product_name = input("请输入产品名称: ").strip()
            # 获取用户输入的最大页数，如果为空则默认50页
            max_pages = int(input("请输入最大页数 (默认50): ") or "50")
            # 执行指定产品的爬取
            crawler.crawl_by_pagination(product_name=product_name, end_page=max_pages)

        elif choice == "3":
            # 选项3：爬取热门蔬菜
            # 定义热门蔬菜列表
            popular_products = ['大白菜', '小白菜', '萝卜', '土豆', '番茄', '黄瓜', '茄子', '辣椒']
            # 爬取热门蔬菜，每个产品最多20页
            crawler.crawl_specific_products(popular_products, max_pages_per_product=20)

        else:
            # 无效选择，使用默认设置
            print("无效选择，默认爬取前10页所有数据")
            crawler.crawl_by_pagination(end_page=10)

    except KeyboardInterrupt:
        # 用户按Ctrl+C中断程序
        print("\n用户中断爬取")
    except Exception as e:
        # 捕获所有其他异常
        logging.error(f"爬取过程中出现错误: {e}")
    finally:
        # 无论如何都要执行的清理工作
        crawler.close()  # 关闭爬虫资源
        print("爬取完成")


# 程序入口点检查
if __name__ == "__main__":
    main()  # 如果直接运行此脚本，则执行main函数